package com.iqiyi.pps.epg.core.service.qipu;

import com.alibaba.fastjson.JSONObject;
import com.iqiyi.kiwi.utils.Configuration;
import com.iqiyi.pps.epg.core.model.channel.ChannelLog;
import com.iqiyi.pps.epg.core.model.program.ProgramData;
import com.iqiyi.pps.epg.core.model.program.ProgramNoticeData;
import com.iqiyi.pps.epg.core.rpc.qipu.QipuService;
import com.iqiyi.pps.epg.core.service.channel.ChannelService;
import com.iqiyi.pps.epg.core.service.program.ProgramService;
import com.iqiyi.pps.epg.core.service.push.AutoPushMsgService;
import com.iqiyi.pps.epg.core.utils.ApplicationContextUtil;
import com.iqiyi.pps.epg.core.utils.Constants;
import com.iqiyi.pps.epg.core.utils.DateTimeUtil;

import com.qiyi.knowledge.rocketmq.client.*;
import com.qiyi.knowledge.rocketmq.client.exception.MQClientException;
import knowledge.pub.ActivemqMessage;
import knowledge.pub.Common;
import knowledge.pub.Metadatas;
import knowledge.pub.ReadService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by karkarrotxu on 2016/6/22.
 */
@Service
@Lazy(false)
public class QipuAsyncServiceImpl implements InitializingBean, IQipuAsyncServiceImpl {
    private static final Logger logger = LoggerFactory.getLogger(QipuAsyncServiceImpl.class);
    public static final String QUEUE_NAME = Configuration.getString("quene.name");
    public static final String ROCKET_MQ_USERNAME = Configuration.getString("rocket.mq.username");
    public static final String ROCKET_MQ_PASSWORD = Configuration.getString("rocket.mq.password");
    public static final String ROCKET_DATA_CENTER = Configuration.getString("rocket.data.center");
    public static final boolean ENABLE_MQ = Configuration.getBoolean("enable.mq");
    public static final String QIPU_LOCK_PRE = "qipu_lock_pre";
    public static final int MAX_RETRY_READ_TIME = Configuration.getInt("max.retry.read.time", 2);
    public static final String KEY_NOTICE_START = "LiveEpisode_StartTime";
    public static final String KEY_NOTICE_END = "LiveEpisode_EndTime";

    public static final String FORMAT_PHASE1 = "修改前 通知开始:%s ; 通知结束: %s\n";
    public static final String FORMAT_PHASE2 = "新数据 通知开始:%s ; 通知结束: %s\n";
    public static final String FORMAT_PHASE3 = "修改后 通知开始:%s ; 通知结束: %s\n";

    private  NotificationMessageConsumer consumer;

    @Autowired
    private QipuService qipuService;

    @Autowired
    private ProgramService programService;

    @Autowired
    private ChannelService channelService;

    public static final long SLEEP_TIME = 200L;

    public void afterPropertiesSet() throws Exception {
        if(!ENABLE_MQ) {
            logger.info("QipuAsyncServiceImpl mq is disabled");
            return;
        }
        if(consumer == null){
            consumer = createConsumer();
            logger.info("QipuAsyncServiceImpl listener  start");
        }
    }

    public NotificationMessageConsumer createConsumer(){
  //      MessageFilter filter = BaseMessageFilter.newBuilder().includeTableList(tableList).includeColumnGroupList(colList).build();
        NotificationMessageConsumer consumer = DefaultNotificationMessageConsumer.newBuilder()
                // 指定消费的环境，请注意该参数值，避免消费错消息
                .dataCenter(DataCenter.valueOf(ROCKET_DATA_CENTER))
                .user(User.newBuilder()
                        // 此处填写申请成功后，奇谱提供的信息（线上与测试信息不同）。
                        .username(ROCKET_MQ_USERNAME)
                        .password(ROCKET_MQ_PASSWORD)
                        .queueName(QUEUE_NAME)
                        .build())
                        // 消费线程池配置
                .threadPoolMinSize(20)  // 最小线程数, 默认20
                .threadPoolMaxSize(64)  // 最大线程数, 默认64
                        // 注册消费消息的listener
                .listener(getListener())
                .build();
        try {
            // 启动消费服务
            // 必须执行该方法才会进行消费，该方法只能调用一次。
            consumer.start();
        } catch (MQClientException e) {
            // 消费者启动失败，请检查错误信息内容，可能是填写信息有误，其他问题请联系奇谱
            logger.error("[createConsumer] error", e);
        }
        return consumer;
    }

    private NotificationMessageListener getListener(){
        return new NotificationMessageListener() {
            @Override
            public void onMessage(ActivemqMessage.NotificationMessage notificationMessage) throws Exception {
                // 处理消息代码
                Long qipuId = null;
                try {
                    qipuId = notificationMessage.getId();
                    Common.EntityIdType entityIdType = com.qiyi.knowledge.common.IdUtil.convertIdToType(qipuId);
                    logger.info("奇谱数据[qipuId={}]", qipuId);
                    if(!Common.EntityIdType.RESOURCE_CONTAINER.equals(entityIdType)){
                        return;
                    }
                    //           LockUtil.accquireLock(QIPU_LOCK_PRE + String.valueOf(qipuId));
                    if (qipuId != null) {
                        doCheckUpdateEntity(qipuId);
                    }
                    logger.info("奇谱数据[qipuId={}] , [session ={}] [start]", qipuId, notificationMessage.getSessionIdList());
                } catch (Exception e) {
                    logger.error("奇谱数据异常 qipuId:" + String.valueOf(qipuId), e);
                }
            }
        };
    }

//    public MessageService doServices() {
//        List<Common.Table> tableList = new ArrayList<Common.Table>();
//        tableList.add(Common.Table.RESOURCE_CONTAINER_TABLE);
//        List<Common.ColumnGroup> colList = new ArrayList<Common.ColumnGroup>();
//        colList.add(Common.ColumnGroup.RESOURCE_ELEMENTS_SUMMARY);
//        MessageFilter filter = BaseMessageFilter.newBuilder().includeTableList(tableList).includeColumnGroupList(colList).build();
//        MessageService service = MessageService.newBuilder()
//                .dataCenter(DataCenter.valueOf(DATA_CENTER.toUpperCase()))
//                .queueName(QUEUE_NAME)
//                .messageFilter(filter).processor(getProcessor()).consumerNumber(5).receiveTimeout(1000).build();
//        try {
//            service.start();
//            logger.info("[doServices] listener init");
//        } catch (Exception e) {
//            logger.error("[doServices] error", e);
//        }
//        return service;
//    }
//    private NotificationMessageProcessor getProcessor() {
//        return new NotificationMessageProcessor() {
//            @Override
//            public void process(ActivemqMessage.NotificationMessage notificationMessage) {
//                Long qipuId = null;
//                try {
//                    qipuId = notificationMessage.getId();
//                    logger.info("奇谱数据[qipuId={}]", qipuId);
//                    if (qipuId % 100 != 12) {
//                        return;
//                    }
//                    //           LockUtil.accquireLock(QIPU_LOCK_PRE + String.valueOf(qipuId));
//                    if (qipuId != null) {
//                        doCheckUpdateEntity(qipuId);
//                    }
//                    logger.info("奇谱数据[qipuId={}] , [session ={}] [start]", qipuId, notificationMessage.getSessionIdList());
//                } catch (Exception e) {
//                    logger.error("奇谱数据异常 qipuId:" + String.valueOf(qipuId), e);
//                } finally {
//                    //              LockUtil.releaseLock(QIPU_LOCK_PRE + String.valueOf(qipuId));
//                }
//            }
//        };
//    }

    private void doCheckUpdateEntity(Long qipuId) {
        logger.info("[doCheckUpdateEntity] start qipuId={}", qipuId);
        try {
            Thread.sleep(SLEEP_TIME);
            ReadService.GetEntityResponse entityResponse = null;
            knowledge.pub.Knowledge.ResourceContainer entity = null;
            int retryTime = 0;
            while (entityResponse == null && retryTime < MAX_RETRY_READ_TIME) {
                entity = (knowledge.pub.Knowledge.ResourceContainer)qipuService.getEntityByqipuId(qipuId);
                retryTime++;
            }
            if (entity == null) {
                logger.error("[doCheckUpdateEntity] fail to read qipuId={}", qipuId);
                return;
            }

            boolean isMobile = false;
            if (entity.getBase() != null && entity.getBase().getPlatformCount() > 0) {
                List<Common.DistributionPlatform> platformList = entity.getBase().getPlatformList();
                for (Common.DistributionPlatform platform : platformList) {
                    if (Common.DistributionPlatform.PHONE_APP_IQIYI.equals(platform)) {
                        isMobile = true;
                        break;
                    }

                }
            }
            if (!isMobile) {
                logger.info("not has PHONE_APP_IQIYI,skip , [pqipuId={}]", qipuId);
                return;
            }
            if (entity.getElementsSummaryCount() > 0) {
                List<ProgramNoticeData> updateList = new ArrayList<ProgramNoticeData>();
                for (Metadatas.ResourceMetaSummary resourceMetaSummary : entity.getElementsSummaryList()) {
                    if (resourceMetaSummary.getEntityId() % 100 == 23) {
                        if (resourceMetaSummary.getKeyValuePairCount() > 0) {
                            ProgramNoticeData updateData = new ProgramNoticeData();
                            updateData.setQipuId(resourceMetaSummary.getEntityId());
                            for (Common.KeyValuePair pair : resourceMetaSummary.getKeyValuePairList()) {
                                if (KEY_NOTICE_START.equals(pair.getName())) {
                                    if (pair.getValue() != null && pair.getValue().length() > 0) {
                                        updateData.setNoticeStartTime(Timestamp.valueOf(pair.getValue()));
                                    }

                                } else if (KEY_NOTICE_END.equals(pair.getName())) {
                                    if (pair.getValue() != null && pair.getValue().length() > 0) {
                                        updateData.setNoticeStopTime(Timestamp.valueOf(pair.getValue()));
                                    }

                                }
                            }
                            logger.info("new date original [{}]: ", JSONObject.toJSONString(updateData));
                            logger.info("new date format [qipuId:{},start={},end={}] ", updateData.getQipuId(), getFormatDateTime(updateData.getNoticeStartTime()), getFormatDateTime(updateData.getNoticeStopTime()));
                            updateList.add(updateData);
                        }
                    }
                }
                if (updateList.size() > 0) {
                    ProgramData programData = null;
                    for (ProgramNoticeData temp : updateList) {
                        try {
                            programData = programService.findProgramDataByQipuId(temp.getQipuId());
                            if (programData == null) {
                                logger.info("[not find data : qipuId={}]", temp.getQipuId());
                                continue;
                            }
                            if (hasDiff(temp, programData)) {
                                StringBuffer sb = new StringBuffer();
                                sb.append("qipuId:" + programData.getQipuId() + ";");
                                sb.append(String.format(FORMAT_PHASE1, getFormatDateTime(programData.getNoticeStartTime()), getFormatDateTime(programData.getNoticeStopTime())));
                                sb.append(String.format(FORMAT_PHASE2, getFormatDateTime(temp.getNoticeStartTime()), getFormatDateTime(temp.getNoticeStopTime())));
                                if (temp.getNoticeStartTime() != null) {
                                    programData.setNoticeStartTime(temp.getNoticeStartTime());
                                }
                                if (temp.getNoticeStopTime() != null) {
                                    programData.setNoticeStopTime(temp.getNoticeStopTime());
                                }
                                sb.append(String.format(FORMAT_PHASE3, getFormatDateTime(programData.getNoticeStartTime()), getFormatDateTime(programData.getNoticeStopTime())));
                                writeChannelLog(programData.getFatherId(), sb.toString());
                                programService.editProgramData(programData);
                                AutoPushMsgService autoPushMsgService = (AutoPushMsgService) ApplicationContextUtil.getBean(AutoPushMsgService.class);
                                autoPushMsgService.updatePushByEpisodeId(programData);
                                programService.doSyncById(programData.getId());
                                logger.info("task diff: [pid={}][qipuId={}]", programData.getId(), temp.getQipuId());
                            }
                        } catch (Exception e) {
                            logger.error("doCheckUpdateEntity cycle", e);
                        }
                    }
                }
            }
            logger.info("[doCheckUpdateEntity] end qipuId={}", qipuId);
        } catch (Exception e) {
            logger.error("[doCheckUpdateEntity] error qipuId=" + qipuId, e);
        }
    }

    private String getFormatDateTime(Timestamp timestamp) {
        return timestamp == null ? "null" : DateTimeUtil.transLongFormat(timestamp.getTime());
    }

    private void writeChannelLog(long channelId, String descr) {
        try {
            ChannelLog channelLog = new ChannelLog();
            channelLog.setFatherId(channelId);
            channelLog.setDescr(descr);
            channelLog.setType(Constants.LogType.QIPU_LISTEN_UPDATE.getIndex());
            channelLog.setUser(Constants.SYSTEM_USER_SIGN);
            channelService.saveLog(channelLog);
        } catch (Exception e) {
            logger.error("writeChannelLog", e);
        }
    }

    private boolean hasDiff(ProgramNoticeData temp, ProgramData programData) {
        boolean diff1 = false;
        boolean diff2 = false;
        if (temp.getNoticeStartTime() != null) {
            diff1 = getTimeValue(temp.getNoticeStartTime()) != getTimeValue(programData.getNoticeStartTime());
        }
        if (temp.getNoticeStopTime() != null) {
            diff2 = getTimeValue(temp.getNoticeStopTime()) != getTimeValue(programData.getNoticeStopTime());
        }
        return diff1 || diff2;
    }

    private long getTimeValue(Timestamp timestamp) {
        return timestamp == null ? 0 : timestamp.getTime();
    }


}
